package drds.binlog.client.adapter.loader;

import drds.binlog.client.adapter.OuterAdapter;
import drds.binlog.client.adapter.support.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 外部适配器的加载器
 */
public class AdapterLoader
{

    private static final Logger logger = LoggerFactory.getLogger(AdapterLoader.class);

    private ClientConfig clientConfig;

    private Map<String, AdapterWorker> canalWorkers = new HashMap<>();

    private Map<String, AdapterKafkaWorker> canalKafkaWorkers = new HashMap<>();

    private ExtensionLoader<OuterAdapter> outerAdapterExtensionLoader;

    public AdapterLoader(ClientConfig clientConfig)
    {
        this.clientConfig = clientConfig;
    }

    /**
     * 初始化canal-client、 canal-client-kafka的适配器
     */
    public void init()
    {
        // canal instances 和 kafka topics 配置不能同时为空
        if (clientConfig.getInstances() == null && clientConfig.getKafkaTopics() == null)
        {
            throw new RuntimeException("Blank config property: canalInstances or canalKafkaTopics");
        }

        outerAdapterExtensionLoader = ExtensionLoader.getExtensionLoader(OuterAdapter.class,
                "" /* TODO clientConfig.getClassloaderPolicy() */);

        String canalServerHost = this.clientConfig.getServerHost();
        SocketAddress socketAddress = null;
        if (canalServerHost != null)
        {
            String[] ipPort = canalServerHost.split(":");
            socketAddress = new InetSocketAddress(ipPort[0], Integer.parseInt(ipPort[1]));
        }
        String zkHosts = this.clientConfig.getZookeeperHosts();

        // if (zkHosts == null && sa == null) {
        // throw new RuntimeException("Blank config property: canalServerHost or
        // zookeeperHosts");
        // }

        // 初始化canal-client的适配器
        if (clientConfig.getInstances() != null)
        {
            for (Instance instance : clientConfig.getInstances())
            {
                List<List<OuterAdapter>> canalOuterAdapterGroups = new ArrayList<>();

                for (AdapterGroup connectorGroup : instance.getAdapterGroups())
                {
                    List<OuterAdapter> outerAdapterList = new ArrayList<>();
                    for (OuterAdapterConfiguration outerAdapterConfiguration : connectorGroup.getOutAdapters())
                    {
                        loadConnector(outerAdapterConfiguration, outerAdapterList);
                    }
                    canalOuterAdapterGroups.add(outerAdapterList);
                }
                AdapterWorker worker;
                if (zkHosts != null)
                {
                    worker = new AdapterWorker(instance.getInstance(), zkHosts, canalOuterAdapterGroups);
                } else
                {
                    worker = new AdapterWorker(instance.getInstance(), socketAddress, canalOuterAdapterGroups);
                }
                canalWorkers.put(instance.getInstance(), worker);
                worker.start();
                logger.info("Start adapter for canal instance: {} succeed", instance.getInstance());
            }
        }

        // 初始化canal-client-kafka的适配器
        if (clientConfig.getKafkaTopics() != null)
        {
            for (KafkaTopic kafkaTopic : clientConfig.getKafkaTopics())
            {
                for (Group group : kafkaTopic.getGroups())
                {
                    List<List<OuterAdapter>> canalOuterAdapterGroups = new ArrayList<>();

                    List<OuterAdapter> outerAdapters = new ArrayList<>();

                    for (OuterAdapterConfiguration config : group.getOutAdapters())
                    {
                        // for (OuterAdapterConfiguration config : adaptor.getOutAdapters()) {
                        loadConnector(config, outerAdapters);
                        // }
                    }
                    canalOuterAdapterGroups.add(outerAdapters);

                    // String zkServers = clientConfig.getZookeeperHosts();
                    AdapterKafkaWorker canalKafkaWorker = new AdapterKafkaWorker(zkHosts,
                            clientConfig.getBootstrapServers(),
                            kafkaTopic.getTopic(),
                            group.getGroupId(),
                            canalOuterAdapterGroups);
                    canalKafkaWorkers.put(kafkaTopic.getTopic() + "-" + group.getGroupId(), canalKafkaWorker);
                    canalKafkaWorker.start();
                    logger.info("Start adapter for canal-client kafka topic: {} succeed",
                            kafkaTopic.getTopic() + "-" + group.getGroupId());
                }
            }
        }
    }

    private void loadConnector(OuterAdapterConfiguration outerAdapterConfiguration, List<OuterAdapter> canalOutConnectors)
    {
        try
        {
            OuterAdapter outerAdapter = outerAdapterExtensionLoader.getExtension(outerAdapterConfiguration.getName());
            ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
            // 替换ClassLoader
            Thread.currentThread().setContextClassLoader(outerAdapter.getClass().getClassLoader());
            outerAdapter.init(outerAdapterConfiguration);
            Thread.currentThread().setContextClassLoader(classLoader);
            canalOutConnectors.add(outerAdapter);
            logger.info("Load canal adapter: {} succeed", outerAdapterConfiguration.getName());
        } catch (Exception e)
        {
            logger.error("Load canal adapter: {} failed", outerAdapterConfiguration.getName(), e);
        }
    }

    /**
     * 销毁所有适配器 为防止canal实例太多造成销毁阻塞, 并行销毁
     */
    public void destroy()
    {
        if (canalWorkers.size() > 0)
        {
            ExecutorService executorService = Executors.newFixedThreadPool(canalWorkers.size());
            for (AdapterWorker adapterWorker : canalWorkers.values())
            {
                final AdapterWorker adapterWorker1 = adapterWorker;
                executorService.submit(new Runnable()
                {

                    @Override
                    public void run()
                    {
                        adapterWorker1.stop();
                    }
                });
            }
            executorService.shutdown();
        }
        if (canalKafkaWorkers.size() > 0)
        {
            ExecutorService executorService = Executors.newFixedThreadPool(canalKafkaWorkers.size());
            for (AdapterKafkaWorker v : canalKafkaWorkers.values())
            {
                final AdapterKafkaWorker cakw = v;
                executorService.submit(new Runnable()
                {

                    @Override
                    public void run()
                    {
                        cakw.stop();
                    }
                });
            }
            executorService.shutdown();
        }
        logger.info("All canal adapters destroyed");
    }
}
